package fox.framework.rpc.client;

import fox.framework.rpc.client.message.RpcClientRequestInterceptor;
import fox.framework.rpc.client.message.request.RpcClientRequest;
import fox.framework.rpc.client.message.request.RpcClientRequestBuilder;
import fox.framework.rpc.client.message.response.ByteSourceResponseEntity;
import fox.framework.rpc.client.message.response.ResponseEntity;
import fox.framework.rpc.client.message.response.RpcClientResponse;
import fox.framework.rpc.client.message.response.RpcClientResponseWrapper;
import fox.framework.rpc.client.node.RootRpcMetadata;
import fox.framework.rpc.client.node.RootRpcNode;
import fox.framework.rpc.client.node.RpcMetadata;
import fox.framework.rpc.client.node.RpcNode;
import fox.framework.rpc.client.node.rule.Rule;
import fox.framework.rpc.common.ExceptionUtils;
import fox.framework.rpc.common.StringUtils;
import fox.framework.rpc.context.definition.RpcDefinition;
import fox.framework.rpc.converter.RpcMessageConverter;
import fox.framework.rpc.endpoint.channel.RpcChannel;
import fox.framework.rpc.endpoint.channel.RpcStatusChannel;
import fox.framework.rpc.exception.RpcNoMetaSpaceException;
import fox.framework.rpc.exception.RpcRuntimeException;

import java.lang.reflect.Type;
import java.util.Collections;
import java.util.List;

/**
 * @author cuichao
 * @Description: DefaultRpcClient
 * @Date: create in 2021/1/8 15:41
 */
public class DefaultRpcClient extends AbstractRpcClient {

    public DefaultRpcClient(String bind, int port) {
        super(bind, port);
    }

    @Override
    protected RpcMetadata buildMetadata(RpcDefinition definition) {
        RootRpcMetadata metadata = new RootRpcMetadata(definition.metaspace());
        metadata.setLazy(definition.isLazy());
        metadata.setRule(getDefaultRuleFactory().getRule());

        RpcMessageConverter converter = StringUtils.isEmpty(definition.MessageConverterName()) ?
                getDefaultConverter() : getConverterByName(definition.MessageConverterName());
        metadata.setMessageConverter(converter);


        return metadata;
    }

    @Override
    protected RpcNode doCreateRpcNode(RpcMetadata metadata, String nodeAddress) {
        return new RootRpcNode(metadata, nodeAddress);
    }


    @Override
    public Object execute(String metaspace, String path, Object[] parameter, Type expectType) throws RpcRuntimeException {
        RpcMetadata metadata = getMetaMap().getOrDefault(metaspace, null);
        if (metadata == null) {
            String msg = String.format("%s  is not be register metaspace.", metaspace);
            throw new RpcNoMetaSpaceException(msg);
        }

        //获取RpcNode节点
        RpcNode node = chooseNode(metadata);
        if (node == null) {
            String msg = String.format("No available RpcNode with %s", path);
            throw new RpcRuntimeException(msg);
        }
        RpcChannel channel = node.getChannel();
        // lazy init
        if (channel == null) {
            connectNode(node);
            channel = node.getChannel();
        }
        //判断该通道是否为状态化通道
        if (channel instanceof RpcStatusChannel) {
            RpcStatusChannel statusChannel = (RpcStatusChannel) channel;
            //如果通道不是存活的
            if (!statusChannel.isAlive()) {
                processUnavailableRpcNode(node);
                channel = node.getChannel();
            }
        }
        //buildRequest
        RpcClientRequest request = buildRpcClientRequest(metaspace, path, parameter);
        //interceptors
        List<RpcClientRequestInterceptor> clientInterceptors = getRpcRequestInterceptor();
        for (RpcClientRequestInterceptor interceptor : clientInterceptors) {
            try {
                interceptor.intercept(request);
            } catch (RpcRuntimeException e) {
                throw e;
            } catch (Exception e) {
                throw ExceptionUtils.RpcRuntimeExceptionWrap(e);
            }
        }
        //execute
        RpcClientResponse response = channel.request(request);
        //process response
        response = processRpcClientResponse(response, expectType);
        if (response.getEntity().ok()) {
            return response.getEntity().getBody();
        }
        throw ExceptionUtils.RpcRuntimeExceptionWrap(response.getEntity().getCause());
    }

    /**
     * 构建Rpc客户端请求
     *
     * @param metaspace
     * @param path
     * @param parameter
     * @return
     */
    protected RpcClientRequest buildRpcClientRequest(String metaspace, String path, Object[] parameter) {
        //buildRequest
        RpcMessageConverter converter = getDefaultConverter();
        RpcClientRequestBuilder builder = new RpcClientRequestBuilder();
        builder.setProtocol(getDefaultProtocol());
        builder.setPath(path);
        builder.setContentType(converter.getConverterName());
        builder.setConverter(converter);
        builder.setRequestEntityBody(parameter);
        RpcClientRequest request = builder.build();
        return request;
    }


    /**
     * 处理由下层API返回的服务端响应
     *
     * @param response
     * @param expectType
     * @return
     */
    protected RpcClientResponse processRpcClientResponse(RpcClientResponse response, Type expectType) {
        ResponseEntity entity = response.getEntity();
        RpcMessageConverter supportConverter = getSupportConverter(response.getProtocol(), response.getContentType());
        if (entity instanceof ByteSourceResponseEntity) {
            response = new ByteSourceDecodingResponseWrapper(response, expectType, supportConverter);
        }
        return response;
    }


    /**
     * 根据元数据空间名获取到Rpc通讯节点
     *
     * @param metadata
     * @return
     */
    @Override
    protected RpcNode chooseNode(RpcMetadata metadata) {
        String metaspace = metadata.metaspace();
        //懒加载且也未被抓取过元数据 #sync
        //或获取后的列表为null且允许空列表的情况下同步像注册中心抓取数据
        //  (nodes.isEmpty() && isAllowFetchIfListEmpty() 是否允许没有可用的RpcNode之后再次向注册中心强制抓取策略暂忽略
        if (metadata.isLazy() && !metadata.isFetched()) {
            if (metadata instanceof RootRpcMetadata) {
                ((RootRpcMetadata) metadata).metaLock();
            }
            //dcl获取每个不用metadata的锁实现同步防止重复调用
            try {
                if (metadata.isLazy() && !metadata.isFetched()) {
                    fetchNodes(Collections.singletonList(metaspace));
                }
            } finally {
                if (metadata instanceof RootRpcMetadata) {
                    ((RootRpcMetadata) metadata).releaseLock();
                }
            }
        }
        //进行choose
        Rule rule = metadata.rule();
        return rule.choose(metadata);
    }


    /**
     * 处理不可用状态的RpcNode
     * RpcNode.RPC_NODE_READY 始化阶段 需要连接远程RpcNode节点
     * RpcNode.RPC_NODE_PAUSE 暂定阶段 注册中心标记为可用但本地可不用节点 暂停node一般由恢复调度线程来恢复
     * RpcNode.RPC_NODE_SHUTDOWN 停止阶段 注册中心标记该节点已经下线 暂时什么都不做 等待注册中心重新上线
     * RpcNode.RPC_NODE_RUNNING 运行阶段 说明channel不可用后没有处理
     *
     * @param node
     */
    private void processUnavailableRpcNode(RpcNode node) throws RpcRuntimeException {
        int state = node.getState();

        switch (state) {
            case RpcNode.RPC_NODE_READY:
                connectNode(node);
            case RpcNode.RPC_NODE_RUNNING:
                //事件监听处理这里暂不处理
            case RpcNode.RPC_NODE_PAUSE:
                //抛异常
            case RpcNode.RPC_NODE_SHUTDOWN:
                //抛异常
            default:
                //nothing to do
        }
    }


    /**
     * 字节源响应包装器
     * 得到可以解析的ResponseEntity
     */
    private class ByteSourceDecodingResponseWrapper extends RpcClientResponseWrapper {
        /**
         * 返回数据的类型
         */
        private Type type;
        /**
         * 对应好的转换器
         */
        private RpcMessageConverter converter;

        public ByteSourceDecodingResponseWrapper(RpcClientResponse response, Type type, RpcMessageConverter converter) {
            super(response);
            this.type = type;
            this.converter = converter;
        }

        @Override
        public ResponseEntity getEntity() {
            ByteSourceResponseEntity entity = ((ByteSourceResponseEntity) super.getEntity());
            return new ByteSourceDecodingResponseEntity(super.getStatus(), entity, converter, type);
        }

    }

    /**
     * 字节源编码
     */
    private class ByteSourceDecodingResponseEntity extends ResponseEntity {
        /**
         * 消息转换器
         */
        private RpcMessageConverter converter;
        /**
         * 字节源响应实体
         */
        private ByteSourceResponseEntity entity;
        /**
         * 返回响应体的类型
         */
        private Type type;
        /**
         * 是否已经解码
         */
        private volatile boolean isEncode = false;

        private Object body;

        private Exception payload;

        public ByteSourceDecodingResponseEntity(int status, ByteSourceResponseEntity entity, RpcMessageConverter converter, Type type) {
            super(entity.getBody(), status);
            this.entity = entity;
            this.type = type;
            this.converter = converter;
        }

        @Override
        public Object getBody() {
            if (!ok()) {
                return null;
            }
            if (!isEncode) {
                byte[] source = entity.getBody();
                if (source == null || source.length == 0) {
                    body = null;
                } else {
                    body = converter.decode(source, type);
                    isEncode = true;
                }

            }
            return body;
        }

        @Override
        public Exception getCause() {
            if (ok()) {
                return null;
            }
            if (!isEncode) {
                byte[] source = entity.getBody();
                if (source == null || source.length == 0) {
                    body = null;
                } else {
                    String msg = converter.decode(source, String.class);
                    payload = new RpcRuntimeException(msg);
                }
            }
            return payload;
        }
    }

}
